package com.fanxuankai.boot.mqbroker.producer;

import cn.hutool.core.exceptions.ExceptionUtil;
import cn.hutool.core.net.NetUtil;
import com.fanxuankai.boot.mqbroker.Event;
import com.fanxuankai.boot.mqbroker.MqBrokerDingTalkClientHelper;
import com.fanxuankai.boot.mqbroker.producer.autoconfigure.MqBrokerProducerProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.Resource;
import java.util.Optional;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.function.Consumer;

/**
 * @author fanxuankai
 */
public abstract class AbstractMqProducer implements MqProducer<MsgSend>, Consumer<Event<String>> {
    private static final Logger LOGGER = LoggerFactory.getLogger(MsgSendServiceImpl.class);
    @Resource
    private MsgSendService msgSendService;
    @Resource
    private ThreadPoolExecutor threadPoolExecutor;
    @Resource
    private MqBrokerProducerProperties properties;
    @Autowired(required = false)
    private MqBrokerDingTalkClientHelper mqBrokerDingTalkClientHelper;

    @Override
    public void produce(MsgSend msg) {
        try {
            Event<String> event = new Event<>();
            event.setGroup(msg.getMqGroup());
            event.setName(msg.getTopic());
            event.setKey(msg.getCode());
            event.setDataKey(msg.getDataKey());
            event.setData(msg.getData());
            event.setEffectTime(msg.getEffectTime());
            Optional.ofNullable(msg.getRetryCount()).ifPresent(event::setRetryCount);
            accept(event);
            LOGGER.info("消息发送成功, topic: {}, code: {}", msg.getTopic(), msg.getCode());
            if (!isPublisherCallback()) {
                if (properties.isAsyncConfirm()) {
                    threadPoolExecutor.execute(() -> msgSendService.success(msg));
                } else {
                    msgSendService.success(msg);
                }
            }
        } catch (Throwable throwable) {
            msg.setCause(ExceptionUtil.stacktraceToString(throwable));
            msgSendService.failure(msg);
            LOGGER.error("消息发送失败, topic: {}, code: {}", msg.getTopic(), msg.getCode(), throwable);
            if (mqBrokerDingTalkClientHelper != null) {
                mqBrokerDingTalkClientHelper.push("消息发送失败", msg.getTopic(), msg.getCode(),
                        msg.getRetry(), NetUtil.getLocalhostStr());
            }
        }
    }
}
